package cn.doitedu.rtdw.dashboard;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job02_每分钟更新累计pvuv {

    public static void main(String[] args) {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射kafka中dwd层的行为日志明细
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "      session_id  string                                 "+
                        "     ,user_name    string                                "+
                        "     ,event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );


        tenv.executeSql(
                "CREATE TABLE dash_02_pvuvse_mysql (                                       \n" +
                        "  window_start timestamp(3)                                       \n" +
                        "  ,window_end   timestamp(3)                                       \n" +
                        "  ,pv           bigint                                             \n" +
                        "  ,uv           bigint                                             \n" +
                        "  ,ses_cnt      bigint                                             \n" +
                        "  ,PRIMARY KEY (window_start,window_end) NOT ENFORCED             \n" +
                        ") with (                                                          \n" +
                        "   'connector' = 'jdbc',                                          \n" +
                        "   'url' = 'jdbc:mysql://doitedu:3306/doit43',                    \n" +
                        "   'table-name' = 'dash_02_pvuvse',                               \n" +
                        "   'username' = 'root'      ,                                     \n" +
                        "   'password' = 'root'                                            \n" +
                        ")                                                                 "
        );



        // 写统计sql
        // 截止到此刻，今天总的pv数，uv数，会话数，并且每1分钟更新一次结果
        tenv.executeSql(
                        " insert into dash_02_pvuvse_mysql   select                                     \n" +
                                " window_start                                                         \n" +
                                ",window_end                                                           \n" +
                                ",count(1) filter(where event_id = 'page_load') as pv                  \n" +
                                ",count(distinct user_id) as uv                 -- uv                  \n" +
                                ",count(distinct session_id) as ses_cnt         -- 会话数               \n" +
                                "from TABLE(                                                              \n" +
                                "    CUMULATE(TABLE dwd_events_kafka,DESCRIPTOR(rt),INTERVAL '1' MINUTE,INTERVAL '24' HOUR )    \n" +
                                ")                                                                     \n" +
                                "GROUP BY                                                              \n" +
                                " window_start,                                                        \n" +
                                " window_end                                                          \n"
        );



    }

}
